package com.atguigu.education.app.dws;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.education.bean.UserBehaviorBean;
import com.atguigu.education.util.ClickHouseUtil;
import com.atguigu.education.util.DateFormatUtil;
import com.atguigu.education.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @Author luyang
 * @create 2022/9/6 14:40
 */
public class DwsUserBehavior {
    public static void main(String[] args) throws Exception {
        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //设置并行度
        env.setParallelism(1);

//        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //TODO 2 设置检查点和状态后端
        /*     //先开启  多长时间去设置一个状态后端 检查点模式一般使用精准一次
        env.enableCheckpointing(5*60*1000L, CheckpointingMode.EXACTLY_ONCE);
        //超时时间
        env.getCheckpointConfig().setCheckpointTimeout(3*60*1000L);
        //最多可以出现几个状态后端
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        //状态保存位置
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        //上面保存位置如果是在hdfs上的话 那么需要用户名
        System.setProperty("HADOOP_USER_NAME","atguigu");
        //开一下状态后端
        env.setStateBackend(new HashMapStateBackend());
        */
        //TODO 3 读取kafka数据
        String topicName = "dwd_traffic_page_log";
        String groupID = "dwd_user_behavior";

        DataStreamSource<String> topicLogStream = env.addSource(KafkaUtil.getKafkaConsumer(topicName, groupID));
        //topicLogStream.print("log>>>>>>");

        //TODO 4 转换
        SingleOutputStreamOperator<JSONObject> mapStream = topicLogStream.map(JSONObject::parseObject);
        //TODO 5 添加水位
        SingleOutputStreamOperator<JSONObject> watermarks = mapStream.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
            @Override
            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                return element.getLong("ts");
            }
        }));

        KeyedStream<JSONObject, String> keyedStream = watermarks.keyBy(ojb -> ojb.getJSONObject("common").getString("mid"));
        //TODO 6 转换数据
        SingleOutputStreamOperator<UserBehaviorBean> flatMapData = keyedStream.flatMap(new RichFlatMapFunction<JSONObject, UserBehaviorBean>() {
            ValueState<String> homeLastTimeState = null;
            ValueState<String> detailLastTimeState = null;

            @Override
            public void open(Configuration parameters) throws Exception {
//添加状态描述，设置ttl状态的存活时间
                ValueStateDescriptor<String> home_last_time = new ValueStateDescriptor<>("home_last_time", String.class);
                home_last_time.enableTimeToLive(StateTtlConfig.newBuilder(org.apache.flink.api.common.time.Time.days(1))
                        .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite).build());
                //设置状态用于存数据的上一条时间
                homeLastTimeState = getRuntimeContext().getState(home_last_time);

                //添加状态描述，设置ttl状态的存活时间
                ValueStateDescriptor<String> detail_last_time = new ValueStateDescriptor<>("detail_last_time", String.class);
                detail_last_time.enableTimeToLive(StateTtlConfig.newBuilder(org.apache.flink.api.common.time.Time.days(1))
                        .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite).build());
                //设置状态用于存数据的上一条时间
                detailLastTimeState = getRuntimeContext().getState(detail_last_time);
            }

            @Override
            public void flatMap(JSONObject value, Collector<UserBehaviorBean> out) throws Exception {
                // 判断数据为首页还是详情页
                String pageID = value.getJSONObject("page").getString("page_id");
                String visitDt = DateFormatUtil.toDate(value.getLong("ts"));
                if ("home".equals(pageID)) {
                    String homeLastVisit = homeLastTimeState.value();
                    if (homeLastVisit == null || !homeLastVisit.equals(visitDt)) {
                        // 首页独立访客
                        homeLastTimeState.update(visitDt);
                        out.collect(new UserBehaviorBean("", "", 1L, 0L, value.getLong("ts")));
                    }
                } else {
                    // 详情数据页独立访客
                    String detailLastVisit = detailLastTimeState.value();
                    if (detailLastVisit == null || !detailLastVisit.equals(visitDt)) {
                        detailLastTimeState.update(visitDt);
                        out.collect(new UserBehaviorBean("", "", 0L, 1L, value.getLong("ts")));
                    }
                }
            }
        });
        // TODO 开窗
        AllWindowedStream<UserBehaviorBean, TimeWindow> windowedStream = flatMapData.windowAll(TumblingEventTimeWindows.of(Time.seconds(10)));

        SingleOutputStreamOperator<UserBehaviorBean> reduceStream = windowedStream.reduce(new ReduceFunction<UserBehaviorBean>() {
            @Override
            //value1是最终数据，向下传递
            //value2是原始数据
            public UserBehaviorBean reduce(UserBehaviorBean value1, UserBehaviorBean value2) throws Exception {
                value1.setHomeUserCount(value1.getHomeUserCount() + value2.getHomeUserCount());
                value1.setCourse_detailUserCount(value1.getCourse_detailUserCount() + value2.getCourse_detailUserCount());
                return value1;
            }
        }, new AllWindowFunction<UserBehaviorBean, UserBehaviorBean, TimeWindow>() {
            @Override
            //得到开窗时间，已经数据库中的ts
            public void apply(TimeWindow window, Iterable<UserBehaviorBean> values, Collector<UserBehaviorBean> out) throws Exception {
                UserBehaviorBean next = values.iterator().next();
                next.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                next.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                next.setTs(System.currentTimeMillis());
                out.collect(next);
            }
        });
        reduceStream.print();
        // TODO 6、写进click house
        reduceStream.addSink(ClickHouseUtil.getClickHouseSinkFunc("insert into dws_user_behavior_window values(?,?,?,?,?)"));

        env.execute();
    }
}
